package cn.edu.flink.scala.tutorial.process

import cn.edu.flink.scala.tutorial.source.{ClickSource, Event}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object KeyedProcessFunctionTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()

    val stream = env.addSource(new ClickSource)

    stream.keyBy(value => true)
      .process(new KeyedProcessFunction[Boolean, Event, String] {
        override def processElement(value: Event, ctx: KeyedProcessFunction[Boolean, Event, String]#Context, out: Collector[String]): Unit = {
          val l = ctx.timerService().currentProcessingTime()
          out.collect("数据到达，当前时间戳：" + l)

          ctx.timerService().registerProcessingTimeTimer(l + 2 * 1000)
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Boolean, Event, String]#OnTimerContext, out: Collector[String]): Unit = {
          out.collect("定时触发，当前时间戳：" + ctx.timerService().currentProcessingTime())
        }
      }).print()

    env.execute("KeyedProcessFunctionTest")
  }
}
